package middleware.p2p;

import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.net.ServerSocket;
import java.net.Socket;

import middleware.p2p.log.MiddlewareLogEvent;
import middleware.p2p.log.MiddlewareMonitorEvent;
import middleware.p2p.log.VectorClock;

import org.apache.log4j.Logger;

/**	
 *	Thread che si occupa di mettersi in ascolto su una porta e di gestire le connessioni
 *	in entrata. Ogni connessione in entrata viene gestita con un thread separato che si occuparà
 *	della deserializzazione del TaskDescriptor e del suo enqueue nella coda.
 *	@see TQueueManagerProxy
 *	@see TQueueManager
 */
public class TaskQueueServerThread extends Thread {
	/**
	 * 	Coda in cui inserire i dati letti dalle connessioni in ingresso.
	 **/
	private TQueueManager queue;
	/**
	 * 	Porta sulla quale rimanere in ascolto per l'arrivo di nuovi TaskDescriptor.
	 */
	private int port;
	/**
	 * 	Classe private implementante un thread che gestisce le connessioni in ingresso.
	 */
	private class TaskQueueRequestHandler extends Thread{
		/**	
		 * 	Coda in cui inserire i TaskDescriptor ricevuti.
		 */
		private TQueueManager tQueue;
		/**
		 * 	Socket dal quale leggere i dati del TaskDescriptor. 
		 */
		private Socket inSock;
		/**
		 * 	Costruttore principale del Thread.
		 * 	@param	queue	Coda nella quale inserire i dati.
		 * 	@param	inSock	Socket dal quale leggere i dati.  
		 */
		public TaskQueueRequestHandler(TQueueManager queue,Socket inSock) {
			this.inSock=inSock;
			this.tQueue=queue;			
		}
		/**
		 * 	Metodo che implementa le operazioni del thread.
		 * 	Il thread gestisce le connessioni in ingresso, leggendo i dati dal socket e inserendoli nella coda.
		 * 	Infine il Thread riceve il file e lo salva nel filesystem.
		 */
		public void run() {
			Logger log=Logger.getLogger(this.getClass());
			try {
				log.debug("Handling connection from "+inSock.getInetAddress().getHostAddress()+":"+inSock.getPort());
				InputStream inSockStream=this.inSock.getInputStream();
				ObjectInputStream inStream=new ObjectInputStream(inSockStream);
				Object item=inStream.readObject();
				if (!(item instanceof MiddlewareMessage)) {
					log.error("Receiving wrong objects from "+inSock.getInetAddress().getHostAddress()+":"+inSock.getPort()+". "+
							"Received "+item.getClass().getName()+" but "+MiddlewareMessage.class.getName()+" expected.");
					return;
				}
				MiddlewareMessage msg=(MiddlewareMessage)item;
				TaskDescriptor newTs=msg.getPayload();
				MiddlewareMonitorEvent monitorEvent=new MiddlewareMonitorEvent();
				this.tQueue.enqueue(newTs,monitorEvent);
				CORE.getLogVectorClock().resyncClocks(msg.getLogVclock(), CORE.getComponentId());
				CORE.getMonitorVectorClock().resyncClocks(msg.getMonitorVclock(), CORE.getComponentId());
				VectorClock logLocalClock=CORE.getLogVectorClock().addEventToClock(CORE.getComponentId());
				VectorClock monitorLocalClock=CORE.getMonitorVectorClock().addEventToClock(CORE.getComponentId());
				log.info(new MiddlewareLogEvent(logLocalClock,
						"Enqueued TaskDescriptor with ID "+newTs.getID()+" received from "+inSock.getInetAddress().getHostAddress()+":"+inSock.getPort()));
				monitorEvent.setVclock(monitorLocalClock);
		        monitorEvent.setComponentID(CORE.getComponentId());
		        Logger.getLogger(CORE.monitorTDQueueName).info(monitorEvent.getQueueContent());
		        Logger.getLogger(CORE.monitorTDQueueStatusName).info(monitorEvent);
		        CORE.getReplicaFrontend().storeReplica(monitorEvent.getQueueContent().toXmlString());
				FileOutputStream outFile=new FileOutputStream(newTs.getLink());
				byte buffer[] = new byte[256];
				int byte_read;
				while((byte_read=inSockStream.read(buffer))!=-1) {
					outFile.write(buffer,0,byte_read);
					outFile.flush();
				}
				logLocalClock=CORE.getLogVectorClock().addEventToClock(CORE.getComponentId());
				log.info(new MiddlewareLogEvent(logLocalClock,
						"File "+newTs.getLink()+" loaded from Socket connected to "+inSock.getInetAddress().getHostAddress()+":"+inSock.getPort()));
				outFile.close();
				inSockStream.close();
				inStream.close();
				this.inSock.close();
			} catch (IOException ioe) {
				ioe.printStackTrace();
				log.error(ioe, ioe);
				System.exit(0);
			} catch (ClassNotFoundException cnfe) {
				cnfe.printStackTrace();
				log.error(cnfe,cnfe);
				System.exit(0);
			}
		}
	}
	/**
	 * 	Costruttore principale del thread
	 * 	@arg	queue	Coda in cui inserire i TaskDescriptor ricevuti.
	 * 	@arg	port	Porta sul quale ascoltare le richieste.
	 */
	public TaskQueueServerThread(TQueueManager queue,int port) {
		this.queue=queue;		
		this.port =port;
	}
	/**
	 * 	Metodo che implementa le operazioni del thread.
	 * 	Il thread si mette in ascolto su una porta e gestisce le connessioni in ingresso.
	 */
	public void run() {
		Logger log=Logger.getLogger(this.getClass());
		ServerSocket serv;
		try {			
			serv = new ServerSocket(this.port);
			log.info("Server socket created on port "+this.port);
		} catch (IOException ioe) {
			ioe.printStackTrace();			
			log.error(ioe,ioe);
			return;
		}		
		while (true) {
			try {
				log.debug("Waiting for incoming connection on "+this.port+" port");
				Socket inSock=serv.accept();
				TaskQueueRequestHandler th = new TaskQueueRequestHandler(this.queue, inSock);
				th.start();
			} catch (IOException ioe) {
				ioe.printStackTrace();
				log.error(ioe,ioe);
				return;
			}
		}
	}	
}
